package com.hz.cheetah.hub.zookeeper;

import com.hz.cheetah.hub.ServiceDiscovery;
import com.hz.cheetah.hub.protocol.CheetahURL;
import org.I0Itec.zkclient.IZkChildListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 基于 ZooKeeper 的服务发现接口实现
 *
 * @author yangchuan
 * @since 1.0.0
 */
public class ZooKeeperServiceDiscovery implements ServiceDiscovery {

    private ZookeeperClient zookeeperClient;
    private final Random random = new Random();

    private static final Logger LOGGER = LoggerFactory.getLogger(ZooKeeperServiceDiscovery.class);

    private String zkAddress;

    private ConcurrentHashMap<String,Set<CheetahURL>> nodes = new ConcurrentHashMap<>();

    final String rootPath = Constant.ZK_ROOT_PATH+ "/";

    public ZooKeeperServiceDiscovery(String zkAddress) {
        this.zkAddress = zkAddress;
    }

    public ZooKeeperServiceDiscovery() {
    }

    public void setZkAddress(String zkAddress) {
        this.zkAddress = zkAddress;
    }

    public void init() {
        zookeeperClient = new ZookeeperClientImp(zkAddress, Constant.ZK_SESSION_TIMEOUT, Constant.ZK_CONNECTION_TIMEOUT);
        zookeeperClient.subscribeChildChanges(Constant.ZK_ROOT_PATH, interfaceListener);
        List<String> interfaces = zookeeperClient.getChildren(Constant.ZK_ROOT_PATH);
        String providers = null;
        for(String service : interfaces){
            providers = rootPath+service+Constant.SERVICE_PROVIDERS;
            if(zookeeperClient.exists(providers)) {
                zookeeperClient.subscribeChildChanges(providers, providesListener);
                resetLocalNodes(service, zookeeperClient.getChildren(providers));
            }
        }
        providers = null;
        interfaces = null;
    }

    /**
     * 监听接口列表，给接口安装提供者监听器，
     * zkclient 的监听数组用的CopyOnWriteArraySet 不会添加重复的对象。故不必担心相同parentPath添加重复的监听器，
     *
     */
    IZkChildListener interfaceListener = new IZkChildListener() {
        @Override
        public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
            Set<String> keys = nodes.keySet();
            String providers = null;
            for(String child : currentChilds){
                if(!keys.contains(child)){
                    providers = parentPath +"/"+child + Constant.SERVICE_PROVIDERS;
                    zookeeperClient.subscribeChildChanges(providers, providesListener);
                    resetLocalNodes(child, zookeeperClient.getChildren(providers));
                }
            }
            providers = null;
            keys = null;

            Set<Map.Entry<String,Set<CheetahURL>>> entrys = nodes.entrySet();
            Iterator<Map.Entry<String,Set<CheetahURL>>> iterator = entrys.iterator();
            String key = null;
            while(iterator.hasNext()){
                key = iterator.next().getKey();
                if(!currentChilds.contains(key)){
                    nodes.remove(key);
                }
            }
            key = null;
            iterator = null;
            entrys = null;
        }
    };


    /**
     * 监听服务接口的提供者
     */
    IZkChildListener providesListener = new IZkChildListener() {
        @Override
        public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
            parentPath = parentPath.substring(Constant.ZK_ROOT_PATH.length()+1,parentPath.length() - Constant.SERVICE_PROVIDERS.length());
            resetLocalNodes(parentPath, currentChilds);
        }
    };

    /**
     * 当节点发生变化时重新更新新的节点
     * 服务器变动不回频繁，所以不用太担心此处性能问题。
     * @param parentPath
     * @param currentChilds
     */
    protected void resetLocalNodes(String parentPath, List<String> currentChilds) {
        Set<CheetahURL> urls = new HashSet<>();
        CheetahURL url = null;
        for(String str : currentChilds){
            url = CheetahURL.pase2CURL(str);
            url.init();
            urls.add(url);
        }
        url = null;
        nodes.put(parentPath,urls);
    }


    @Override
    public CheetahURL discover(String name) {
        return discover(name,nodes.get(name));
    }


    /**
     * 找到服务器，支持按权重的负载均衡算法
     * @param name
     * @param urlStrs
     * @return
     */
    public CheetahURL discover(String name,Set<CheetahURL> urlStrs) {
        if (urlStrs == null || urlStrs.size() == 0) {
            throw new RuntimeException(String.format("can not find any address node on path: %s", name));
        }
        if (urlStrs.size() == 1) {
            CheetahURL url = urlStrs.iterator().next();
            return url;
        } else {
            Iterator<CheetahURL> iterator = urlStrs.iterator();
            int totalWeight = 0;
            while (iterator.hasNext()) {
                totalWeight += iterator.next().getWeight();
            }
            int offset = random.nextInt(totalWeight);
            CheetahURL targetUrl = null;
            iterator = urlStrs.iterator();
            while (iterator.hasNext()) {
                targetUrl = iterator.next();
                offset -= targetUrl.getWeight();
                if (offset < 0) {
                    break;
                }
            }
            iterator = null;
            return targetUrl;
        }
    }

    /**
     * 当第一次发现的服务器有问题后，重新发现发现一台服务器
     * @param serviceName
     * @param old
     * @return
     */
    public CheetahURL reDiscover(String serviceName,CheetahURL old) {
        Set<CheetahURL> urls = nodes.get(serviceName);
        HashSet<CheetahURL> newUlrs = new HashSet();
        Iterator<CheetahURL> iterator = urls.iterator();
        while(iterator.hasNext()){
            newUlrs.add(iterator.next());
        }
        if(newUlrs.size() > 1) {
            newUlrs.remove(old);
        }
        iterator = null;
        urls = null;

        return discover(serviceName,newUlrs);
    }
}